package com.ylkj.bus.dispatcher.handler;

import com.ylkj.bus.MessageContent;
import com.ylkj.bus.app.report.DeviceContainerDO;
import com.ylkj.bus.app.report.UnpackIotpdeviceConDeleteAndInsert;
import com.ylkj.bus.container.report.ReportContainerProcess;
import com.ylkj.bus.dispatcher.IStreamHandler;
import com.ylkj.dao.RdsConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.util.OutputTag;

import java.util.List;

/**
 * @Author: wengzx
 * @Date: 2025/3/28
 * @Desc:
 */
@RequiredArgsConstructor
@Slf4j
public class ReportContainerStreamHandler implements IStreamHandler {
    private final RdsConfig rdsConfig;
    private final String tableName;

    @Override
    public void handle(DataStream<MessageContent> stream) {
        // 定义侧输出流标签
        OutputTag<List<DeviceContainerDO>> outTagRds1 = new OutputTag<List<DeviceContainerDO>>("outTagRds1") {};

        SingleOutputStreamOperator<MessageContent> reportContainerProcess = stream.process(new ReportContainerProcess(outTagRds1)).name("report_container_process");
        DataStream<List<DeviceContainerDO>> sideOutput = reportContainerProcess.getSideOutput(outTagRds1);
        //sideOutput.print();
        // 将侧输出流传递给处理类
        sideOutput.process(new UnpackIotpdeviceConDeleteAndInsert(tableName, rdsConfig)).name("unpack_iotp_device_container_delete_and_insert");
    }
}
